package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

//ProcessFunction<IN, OUT>
public class AppLogSplitProcessFunctionlx extends ProcessFunction<String, String> {

    private final OutputTag<String> startTag;
    private final OutputTag<String> errorTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public AppLogSplitProcessFunctionlx(OutputTag<String> startTag, OutputTag<String> errorTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.startTag = startTag;
        this.errorTag = errorTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
        JSONObject jsonObject = JSON.parseObject(s);


        if (null != jsonObject.get("err")) {
            context.output(errorTag, s);
            jsonObject.remove("err");
        }

        if (null != jsonObject.get("start")) {
            context.output(startTag, s);
        } else {
            JSONArray dispalysJsonArray = jsonObject.getJSONArray("displays");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            // 输出页面日志
            collector.collect(jsonObject.toJSONString());

            if(null!=dispalysJsonArray && !dispalysJsonArray.isEmpty()){
                for (int i = 0; i < dispalysJsonArray.size(); i++) {
                    JSONObject displayJsonObject = dispalysJsonArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("display", dispalysJsonArray);
                    //输出
                    context.output(displayTag, jsonObject.toJSONString());
                }
            }
            jsonObject.remove("display");

            if(null!=actionsJsonArray && !actionsJsonArray.isEmpty()){
                for (int i = 0; i < actionsJsonArray.size(); i++) {
                    JSONObject actionJsonObject = actionsJsonArray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("action", actionsJsonArray);
                    //输出
                    context.output(actionTag, jsonObject.toJSONString());
                }
            }





        }


    }
}
